package com.zhida;

import okhttp3.ResponseBody;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.select.Elements;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rx.Observable;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.schedulers.Schedulers;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by Administrator on 2017-04-05.
 */
public class Spider {

    private final static Logger logger = LoggerFactory.getLogger(Spider.class);


    private BlockingQueue<String> queue = new LinkedBlockingQueue<String>();

    private ListProcessor listProcessor = new ListProcessor();

    private final AtomicBoolean gettingList = new AtomicBoolean(false);
    private final AtomicInteger runningCount = new AtomicInteger(0);

    public void init() {
        /*Disposable stats = Flowable.interval(1, 2, TimeUnit.SECONDS)
                .subscribe(new Consumer<Long>() {
                    public void accept(@NonNull Long aLong) throws Exception {
                        logger.info("当前正在解析的进程有 {} 个, 还有 {} 个url 等待被解析", runningCount.get(), queue.size());
                    }
                });*/


        /*Observable.create(new Observable.OnSubscribe<String>() {
            public void call(Subscriber<? super String> subscriber) {
                while (true) {
                    String url = queue.poll();
                    if (url != null) {
                        subscriber.onNext(url);
                    } else {
                        page++;
                        System.out.println("去解析更多url………………");
                        getList();
                        try {
                            Thread.sleep(3000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        })
                .doOnNext(new Action1<String>() {
                    public void call(String s) {
                        runningCount.incrementAndGet();
                    }
                })
                .flatMap(new Func1<String, Observable<?>>() {
                    public Observable<?> call(final String url) {
                        return RetrofitFactory.getGoOutService().get(RetrofitFactory.BASE_URL + url)
                                .subscribeOn(Schedulers.io())
                                .flatMap(new Func1<ResponseBody, Observable<?>>() {
                                    public Observable<?> call(ResponseBody responseBody) {
                                        try {
                                            String html = responseBody.string();
                                            System.out.println("完成解析" + url);
                                            return Observable.just(html);
                                        } catch (IOException e) {
                                            e.printStackTrace();
                                        }
                                        return Observable.never();
                                    }
                                }, 2);
                    }
                })
                .doOnNext(new Action1<Object>() {
                    public void call(Object o) {
                        runningCount.decrementAndGet();
                    }
                })
                .subscribe(new Action1<Object>() {
                    public void call(Object o) {
//                        System.out.println("任务完成");
                    }
                });*/

        /*for (int i = 0; i < 2; i++) {
            start();
        }*/
        start();

    }

    public void start() {

        String url = queue.poll();
        if (url != null) {
            synchronized (Spider.class) {
                int current = runningCount.get();
                if (current < 2) {
                    for (int i = 0; i < 2 - current; i++) {
                        handleContent(url);
                    }
                }
            }
        } else {
            getList();
        }


        /*Observable.create(new Observable.OnSubscribe<String>() {
            public void call(Subscriber<? super String> subscriber) {
                String url = queue.poll();
                while (true) {
                    if (url != null) {
                        subscriber.onNext(url);
                    } else {
                        page++;
                        getList();
                    }
                }
            }
        })
                .doOnNext(new Action1<String>() {
                    public void call(String s) {
                        runningCount.incrementAndGet();
                    }
                })
                .flatMap(new Func1<String, Observable<?>>() {
                    public Observable<?> call(String s) {
                        queue.offer(s);
                        return Observable.just(1, 2, 3);
                    }
                }, 5)
                .doOnNext(new Action1<Object>() {
                    public void call(Object o) {
                        runningCount.decrementAndGet();
                    }
                })
                .subscribe(new Action1<Object>() {
                    public void call(Object o) {

                    }
                });*/

                /*.subscribe(new Subscriber<String>() {
                    public void onCompleted() {
                        System.out.println("onCompleted");
                    }

                    public void onError(Throwable e) {

                    }

                    public void onNext(String s) {
                        System.out.println(s);
                    }
                });*/
    }

    public void onContentFinish() {
        start();
    }

    public void handleContent(final String url) {
        runningCount.incrementAndGet();
        RetrofitFactory.getGoOutService().get(RetrofitFactory.BASE_URL + url)
                .subscribeOn(Schedulers.io())
                .flatMap(new Func1<ResponseBody, Observable<?>>() {
                    public Observable<?> call(ResponseBody responseBody) {
                        try {
                            String html = responseBody.string();
                            Document doc = Jsoup.parse(html);
                            Elements elements = doc.select("body");

                            return Observable.just(html);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                        return Observable.never();
                    }
                })
                .subscribe(new Action1<Object>() {
                    public void call(Object o) {
                        runningCount.decrementAndGet();
                        logger.info("完成解析" + url);
                        onContentFinish();
                    }
                });
    }

    private int page = 1;

    public void onListFinish() {
        start();
    }

    public void getList() {

        if (gettingList.get()) {
            return;
        }
        gettingList.set(true);

        IGoOutService goOutService = RetrofitFactory.getGoOutService();


        String url;
        if (page == 1) {
            url = RetrofitFactory.BASE_URL + "article/fwydyl/zgzx/?";
        } else {
            url = RetrofitFactory.BASE_URL + "/" + "article/fwydyl/zgzx/?" + page;
        }

        goOutService.get(url)
                .subscribeOn(Schedulers.io())
                .subscribe(new Action1<ResponseBody>() {
                    public void call(ResponseBody responseBody) {
                        try {
                            String html = responseBody.string();

                            List<String> urls = listProcessor.parse(html);
                            queue.addAll(urls);

                            gettingList.set(false);

                            onListFinish();
                            System.out.println(queue);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }, new Action1<Throwable>() {
                    public void call(Throwable throwable) {
                        throwable.printStackTrace();
                    }
                });

    }
}
